package netbolt

import (
	"errors"
	"flag"
	"github.com/samuel/go-zookeeper/zk"
	"io/ioutil"
	"os"
	"strconv"
	"sync"
)

var (
	myId               string = ""
	onlineNodeWatch    <-chan zk.Event
	clusterConfigWatch <-chan zk.Event
	configVersion      int
	onlines            = map[string]*DataNode{}
	offlines           = map[string]*DataNode{}
	hostStatusLock     sync.RWMutex
	NodeSize           int //应有的节点数

	NOT_FOUND_CLUSTER_ERR = errors.New("Not found cluster err")
)

type DataNode struct {
	Min           uint16
	Max           uint16
	MyId          string
	Host          string
	Port          uint16
	HasMasterNode bool
}

func readMyId() string {
	f, err := os.Open(NetBoltConfig.Netbolt.DBPath + "/myid")
	if err != nil {
		ErrLog.Panic(err)
	}
	defer f.Close()

	buf, err := ioutil.ReadAll(f)
	if err != nil {
		ErrLog.Panic(err)
	}

	return string(buf)
}

//默认情况下不能对配置进行修改，只能增加主机和减少主机在这种情况下会做特殊处理。
func InitCloud() {
	if flag.Lookup("test.v") == nil && NetBoltConfig.Netbolt.Cluster {
		if _, err := ZKCreateIfNotExist(ZK_STATUS_PATH, nil, 0, zk.WorldACL(zk.PermAll)); err != nil {
			ErrLog.Panic(err)
		}

		if _, err := ZKCreateIfNotExist(ZK_CLUSTER_PATH, nil, 0, zk.WorldACL(zk.PermAll)); err != nil {
			ErrLog.Panic(err)
		}
		if _, err := ZKCreateIfNotExist(ZK_CLUSTER_COMPLETE_PATH, nil, 0, zk.WorldACL(zk.PermAll)); err != nil {
			ErrLog.Panic(err)
		}
		if _, err := ZKCreateIfNotExist(ZK_ONLINE_PATH, nil, 0, zk.WorldACL(zk.PermAll)); err != nil {
			ErrLog.Panic(err)
		}

		if exist, _, _ := ZkConn.Exists(ZK_ONLINE_PATH + "/" + myId); !exist {
			ZkConn.Create(ZK_ONLINE_PATH+"/"+myId, []byte(NetBoltConfig.Netbolt.Listen), zk.FlagEphemeral, zk.WorldACL(zk.PermAll))
		} else {

			err := ZkConn.Delete(ZK_ONLINE_PATH+"/"+myId, -1)
			if err != nil {
				WARNLog.Println(err)
			}
			ZkConn.Create(ZK_ONLINE_PATH+"/"+myId, []byte(NetBoltConfig.Netbolt.Listen), zk.FlagEphemeral, zk.WorldACL(zk.PermAll))

		}

		_, _, onlineNodeWatch, err := ZkConn.ChildrenW(ZK_ONLINE_PATH)

		paths, _, clusterConfigWatch, err := ZkConn.ChildrenW(ZK_CLUSTER_COMPLETE_PATH)
		if err != nil {
			ErrLog.Panic(err)
		}

		if len(paths) == 0 {
			configVersion = -1 //未初始化状态
		} else {
			configVersion, _ = strconv.Atoi(paths[0][1:])

		}

		Logger.Println("Current cluster config version:", configVersion)

		if configVersion == -1 {
			Logger.Println("Wait config cluster !")
		} else {
			InitDataNode()
		}

		//配置监控
		go func() {
			for {
				select {
				case e := <-clusterConfigWatch:
					//if e.State == zk.StateDisconnected {
					//
					//	exist, _, err := ZkConn.Exists(ZK_CLUSTER_COMPLETE_PATH)
					//	if err != nil {
					//		ErrLog.Panic(err)
					//	}
					//	if exist {
					//		_, _, clusterConfigWatch, err = ZkConn.ChildrenW(ZK_CLUSTER_COMPLETE_PATH)
					//		if err != nil {
					//			ErrLog.Panic(err)
					//		}
					//	}
					//
					//} else {
					if DEBUG {
						Logger.Printf("zk event type:%s,state:%s,err:%v,path:%s,server:%s\n", e.Type.String(), e.State.String(), e.Err, e.Path, e.Server)
					}

					if e.Type == zk.EventNodeChildrenChanged {
						paths, _, clusterConfigWatch, err = ZkConn.ChildrenW(ZK_CLUSTER_COMPLETE_PATH)
						if err != nil {
							ErrLog.Panic(err)
						}

						if len(paths) > 0 {
							newConfigVersion, _ := strconv.Atoi(paths[0][1:])

							if newConfigVersion > configVersion {
								configVersion = newConfigVersion

								InitDataNode()
							}
							Logger.Println("Current cluster config version:", configVersion)

						}
					}
					//}
				}
			}
		}()

		//监控主机
		go func() {
			for {
				select {
				case e := <-onlineNodeWatch:

					if DEBUG {
						Logger.Printf("zk event type:%s,state:%s,err:%v,path:%s,server:%s\n", e.Type.String(), e.State.String(), e.Err, e.Path, e.Server)
					}
					if e.Type == zk.EventNodeChildrenChanged {
						paths, _, onlineNodeWatch, err = ZkConn.ChildrenW(ZK_ONLINE_PATH)

						InitDataNode()

					}

				}
			}
		}()

	}
}

//判断所有数据节点是否完备
func Whole() bool {
	if configVersion <= -1 {
		return false
	}
	hostStatusLock.RLock()
	defer hostStatusLock.RUnlock()
	return len(onlines) == NodeSize
}

func InitDataNode() error {

	hostStatusLock.Lock()
	defer hostStatusLock.Unlock()
	if configVersion == -1 {
		return nil
	}
	paths, _, err := ZkConn.Children(ZK_CLUSTER_PATH)
	NodeSize = len(paths)
	if err != nil {
		ErrLog.Panic(err)
	}

	if len(paths) == 0 {
		return NOT_FOUND_CLUSTER_ERR
	}

	for _, path := range paths {
		buf, _, err := ZkConn.Get(ZK_CLUSTER_PATH + "/" + path)

		if err != nil {
			ErrLog.Panic(err)
		}
		dn := &DataNode{}
		err = DataSerializer.FromBytes(buf, dn)

		if err != nil {
			ErrLog.Panic(err)
		}

		exist, _, err := ZkConn.Exists(ZK_ONLINE_PATH + "/" + dn.MyId)
		if err != nil {
			ErrLog.Panic(err)
		}
		if exist {
			onlines[dn.MyId] = dn
			Logger.Println("Online ", dn)
		} else {
			offlines[dn.MyId] = dn
			Logger.Println("Offline ", dn)
		}

	}

	return nil
}

func SyncKey(keys [][]byte) {

}
